package com.demo.util;

import cn.hutool.crypto.SecureUtil;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import java.util.function.Predicate;

/**
 * @author wangfengchen
 */
@Slf4j
@Component
public class RedisDelayedQueueUtil {

  @Autowired
  private RedissonClient redissonClient;
  @Autowired
  private ThreadPoolTaskExecutor threadPoolTaskExecutor;

  public final static String QUEUE_NAME = "DELAY_QUEUE";

  private final boolean isReSend = false;
  private final Long[] RESEND_DELAYED_SECONDS = {TimeUnit.SECONDS.toSeconds(15)
      , TimeUnit.MINUTES.toSeconds(1), TimeUnit.MINUTES.toSeconds(10)
      , TimeUnit.HOURS.toSeconds(1), TimeUnit.HOURS.toSeconds(5)};
  private final Map<String, AtomicInteger> MSG_RETRY_COUNTER = new ConcurrentHashMap<>();


  /**
   * 添加消息到延迟队列
   *
   * @param queueName 队列名称
   * @param msg       消息
   * @param removeOld 是否删除旧消息
   * @param delay     延迟时间
   * @param timeUnit  时间单位
   */
  public <T> void addDelayQueue(@NonNull String queueName, @NonNull T msg, boolean removeOld, long delay, TimeUnit timeUnit) {
    RBlockingDeque<T> blockingDeque = redissonClient.getBlockingDeque(queueName);
    RDelayedQueue<T> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
    if (removeOld) {
      delayedQueue.remove(msg);
    }
    log.info("【队列-{}】- 发送消息 = {}，delay = {}", queueName, msg, delay);
    delayedQueue.offer(msg, delay, timeUnit);
  }

  /**
   * 添加消息到延迟队列
   *
   * @param queueName 队列名称
   * @param msg       消息
   * @param delay     延迟时间
   * @param timeUnit  时间单位
   */
  public <T> void addDelayQueue(@NonNull String queueName, @NonNull T msg, long delay, TimeUnit timeUnit) {
    addDelayQueue(queueName, msg, false, delay, timeUnit);
  }

  /**
   * 删除指定任务
   */
  public <T> void removeTask(@NonNull String queueName, @NonNull T msg) {
    RBlockingDeque<T> blockingDeque = redissonClient.getBlockingDeque(queueName);
    RDelayedQueue<T> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
    delayedQueue.remove(msg);
  }

  /**
   * 删除所有指定任务
   */
  public <T> void removeAllTask(@NonNull String queueName, @NonNull T msg) {
    RBlockingDeque<T> blockingDeque = redissonClient.getBlockingDeque(queueName);
    RDelayedQueue<T> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
    Predicate<T> predicate = x -> x.equals(msg);
    delayedQueue.removeIf(predicate);
  }

  /**
   * 清空延迟队列
   */
  public <T> void clearDelayQueue(@NonNull String queueName) {
    RBlockingDeque<T> blockingDeque = redissonClient.getBlockingDeque(queueName);
    RDelayedQueue<T> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
    delayedQueue.clear();
  }

  /**
   * 消费队列
   *
   * @param queueName
   * @param handler
   * @param <T>
   */
  public <T> void consumeQueueMsg(@NonNull String queueName, Consumer<T> handler) {
    threadPoolTaskExecutor.submit(() -> {
      RBlockingDeque<T> blockingDeque = redissonClient.getBlockingDeque(queueName);
      // 注意虽然delayedQueue在这个方法里面没有用到，但是这行代码也是必不可少的
      RDelayedQueue<T> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
      log.info("【队列-{}】- 监听队列成功", queueName);
      while (true) {
        T msg = null;
        try {
          msg = blockingDeque.take();
          handler.accept(msg);
          log.info("【队列-{}】- 处理消息成功，msg = {}", queueName, msg);
        } catch (InterruptedException e) {
          break;
        } catch (Exception e) {
          log.error("【队列-{}】- 处理消息失败，msg = {}", queueName, msg, e);
          // 重试机制
          if (isReSend && ObjectUtils.isNotEmpty(msg)) {
            String md5 = SecureUtil.md5(msg.toString());
            AtomicInteger atomicInteger = Optional
                .ofNullable(MSG_RETRY_COUNTER.get(md5))
                .orElseGet(() -> {
                  AtomicInteger count = new AtomicInteger(0);
                  MSG_RETRY_COUNTER.put(md5, count);
                  return count;
                });
            int index = Math.min(atomicInteger.getAndIncrement(), RESEND_DELAYED_SECONDS.length - 1);
            addDelayQueue(queueName, msg, RESEND_DELAYED_SECONDS[index], TimeUnit.SECONDS);
          }
        }
      }
    });
  }

}
